package operator.filter;


import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

class SampleOperator {

    public static void main(String[] args) {
//        sample1();
//        sample2();
        sample3();
    }

    /**
     * 定期采样，采最近发射的数据
     */
    static void sample1() {
        Observable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doOnNext aLong = " + aLong);
                    }
                })
                .sample(3, TimeUnit.SECONDS)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 每当sample中的Observable发出事件，采最近发射的数据
     */
    static void sample2() {
        Observable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doOnNext aLong = " + aLong);
                    }
                })
                .sample(Observable.interval(3, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- sample doOnNext aLong = " + aLong);
                    }
                }))
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 每当sample中的Observable发出事件，采最近发射的数据
     * emitLast 是否取sample结束后的事件
     */
    static void sample3() {
        Observable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("---------------- doOnNext aLong = " + aLong);
                    }
                })
                .sample(Observable.intervalRange(1, 3, 2, 3, TimeUnit.SECONDS)
                        .doOnNext(new Consumer<Long>() {
                            @Override
                            public void accept(Long aLong) throws Exception {
                                Printer.print("---------------- sample doOnNext aLong = " + aLong);
                            }
                        }), true)
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
